package com.zidongxiangxi.sharding.job.starter.processor;

import com.zidongxiangxi.sharding.job.core.ShardingJob;
import com.zidongxiangxi.sharding.job.core.ShardingJobConfig;
import com.zidongxiangxi.sharding.job.core.executor.runnable.ShardingJobRunnable;
import com.zidongxiangxi.sharding.job.core.registry.ZookeeperConfiguration;
import com.zidongxiangxi.sharding.job.core.registry.ZookeeperRegistryCenter;
import com.zidongxiangxi.sharding.job.starter.annotation.ShardingJobRunner;
import com.zidongxiangxi.sharding.job.starter.config.ShardingJobSpringConfig;
import com.zidongxiangxi.sharding.job.starter.config.ShardingJobSpringZookeeperConfig;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.config.InstantiationAwareBeanPostProcessorAdapter;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.PriorityOrdered;
import org.springframework.util.StringUtils;

import java.util.Arrays;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

/**
 * ShardingJobRunner注解bean处理类
 *
 * @author chenxudong
 * @date 2021/02/22
 */
public class ShardingJobRunnerAnnotationBeanPostProcessor extends InstantiationAwareBeanPostProcessorAdapter
    implements ApplicationContextAware, PriorityOrdered, DisposableBean {
    private ApplicationContext applicationContext = null;
    private ZookeeperRegistryCenter registryCenter = null;
    private final Map<String, ShardingJob> jobRunnableCaches = new ConcurrentHashMap<>();

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        if (!(bean instanceof ShardingJobRunnable)) {
            return bean;
        }
        ShardingJobRunner runner = bean.getClass().getAnnotation(ShardingJobRunner.class);
        if (Objects.isNull(runner)) {
            return bean;
        }
        if (StringUtils.isEmpty(runner.jobName()) || runner.shardingItems().length < 1) {
            throw new RuntimeException("任务名称和任务分片项不能为空");
        }
        if (jobRunnableCaches.containsKey(runner.jobName())) {
            throw new RuntimeException("存在重复的任务名");
        }

        if (Objects.isNull(registryCenter)) {
            ShardingJobSpringConfig config = applicationContext.getBean(ShardingJobSpringConfig.class);
            ShardingJobSpringZookeeperConfig zookeeperConfig = config.getZookeeper();
            ZookeeperConfiguration zookeeperConfiguration =
                new ZookeeperConfiguration(zookeeperConfig.getAddress(), config.getApplicationName());
            zookeeperConfiguration.setBaseSleepTimeMilliseconds(zookeeperConfig.getBaseSleepTimeMilliseconds());
            zookeeperConfiguration.setMaxSleepTimeMilliseconds(zookeeperConfig.getMaxSleepTimeMilliseconds());
            zookeeperConfiguration.setMaxRetries(zookeeperConfig.getMaxRetries());
            zookeeperConfiguration.setSessionTimeoutMilliseconds(zookeeperConfig.getSessionTimeoutMilliseconds());
            zookeeperConfiguration.setConnectionTimeoutMilliseconds(zookeeperConfig.getConnectionTimeoutMilliseconds());
            registryCenter = new ZookeeperRegistryCenter(zookeeperConfiguration);
            registryCenter.init();
        }

        ShardingJobConfig shardingJobConfig = new ShardingJobConfig();
        shardingJobConfig.setJobName(runner.jobName());
        shardingJobConfig.setShardingTotalCount(runner.shardingItems().length);
        shardingJobConfig.setShardingItems(Arrays.asList(runner.shardingItems()));
        shardingJobConfig.setOverwrite(runner.override());
        ShardingJob shardingJob = new ShardingJob(shardingJobConfig, registryCenter, (ShardingJobRunnable)bean);
        jobRunnableCaches.put(runner.jobName(), shardingJob);

        shardingJob.start();
        return bean;
    }

    @Override
    public int getOrder() {
        return LOWEST_PRECEDENCE;
    }

    @Override
    public void destroy() {
        for (ShardingJob shardingJob : jobRunnableCaches.values()) {
            shardingJob.shutdown();
        }
        if (Objects.nonNull(registryCenter)) {
            registryCenter.close();
        }
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}
